package connector

import (
	"sync"
)

// ConnectorRegistry represents a connector registry
type ConnectorRegistry struct {
	// factory is the connector factory
	factory *DefaultConnectorFactory
	// mu is the mutex
	mu sync.RWMutex
}

// NewConnectorRegistry creates a new connector registry
func NewConnectorRegistry() *ConnectorRegistry {
	return &ConnectorRegistry{
		factory: NewDefaultConnectorFactory(),
	}
}

// RegisterSourceConnector registers a source connector
func (r *ConnectorRegistry) RegisterSourceConnector(connectorType string, factory func() SourceConnector) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.factory.RegisterSourceConnector(connectorType, factory)
}

// RegisterSinkConnector registers a sink connector
func (r *ConnectorRegistry) RegisterSinkConnector(connectorType string, factory func() SinkConnector) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.factory.RegisterSinkConnector(connectorType, factory)
}

// GetFactory gets the connector factory
func (r *ConnectorRegistry) GetFactory() ConnectorFactory {
	return r.factory
}

// RegisterBuiltinConnectors registers builtin connectors
func (r *ConnectorRegistry) RegisterBuiltinConnectors() {
	// Register core source connectors
	r.RegisterSourceConnector("memory", NewMemorySourceConnector)
	r.RegisterSourceConnector("file", NewFileSourceConnector)
	r.RegisterSourceConnector("csv", NewCSVSourceConnector)
	r.RegisterSourceConnector("parquet", NewParquetSourceConnector)
	r.RegisterSourceConnector("http", NewHTTPSourceConnector)
	r.RegisterSourceConnector("mysql", NewMySQLSourceConnector)

	// Register core sink connectors
	r.RegisterSinkConnector("memory", NewMemorySinkConnector)
	r.RegisterSinkConnector("file", NewFileSinkConnector)
	r.RegisterSinkConnector("csv", NewCSVSinkConnector)
	r.RegisterSinkConnector("parquet", NewParquetSinkConnector)
	r.RegisterSinkConnector("console", NewConsoleSinkConnector)
	r.RegisterSinkConnector("http", NewHTTPSinkConnector)
	r.RegisterSinkConnector("mysql", NewMySQLSinkConnector)

	// Register optional connectors if available
	r.registerOptionalConnectors()
}

// registerOptionalConnectors registers connectors that depend on external packages
func (r *ConnectorRegistry) registerOptionalConnectors() {
	// These connectors require external dependencies and are registered only if available
	// They are implemented in separate files with build tags

	// Register Kafka connectors if available
	// r.RegisterSourceConnector("kafka", NewKafkaSourceConnector)
	// r.RegisterSinkConnector("kafka", NewKafkaSinkConnector)

	// Register MQTT connectors if available
	// r.RegisterSourceConnector("mqtt", NewMQTTSourceConnector)
	// r.RegisterSinkConnector("mqtt", NewMQTTSinkConnector)

	// Register Database connectors if available
	// r.RegisterSourceConnector("database", NewDatabaseSourceConnector)
	// r.RegisterSinkConnector("database", NewDatabaseSinkConnector)
}

// DefaultRegistry is the default connector registry
var DefaultRegistry = NewConnectorRegistry()

// init initializes the default registry
func init() {
	DefaultRegistry.RegisterBuiltinConnectors()
}
